package com.zhss.dfs.namenode.server;

import com.zhss.dfs.namenode.server.context.DfsEnv;
import com.zhss.dfs.namenode.server.editlog.EditLog;
import com.zhss.dfs.namenode.server.editlog.FSEditlog;

import java.io.File;
import java.io.FileOutputStream;
import java.io.RandomAccessFile;
import java.nio.ByteBuffer;
import java.nio.channels.FileChannel;

/**
 * 负责管理元数据的核心组件
 *
 * @author zhonghuashishan
 */
public class FSNameSystem {

    /**
     * 负责管理内存文件目录树的组件
     */
    private FSDirectory directory;
    /**
     * 负责管理edits log写入磁盘的组件
     */
    private FSEditlog editlog;

    public FSNameSystem() {
        this.directory = new FSDirectory();
        this.editlog = new FSEditlog(this);
    }

    private Long checkpointTxid;
    private DfsEnv env = DfsEnv.getInstance();

    /**
     * 创建目录
     *
     * @param path 目录路径
     * @return 是否成功
     */
    public Boolean mkdir(String path) throws Exception {
        this.directory.mkdir(path);
        EditLog log = new EditLog("mkdir", path);
        this.editlog.logEdit(log);
        return true;
    }

    //强制把内存里的editlog刷入磁盘中
    public void flush() {
        editlog.flush();
    }

    //获取fseditlog组件
    public FSEditlog getEditlog() {
        return editlog;
    }

    public void setCheckpointTxid(long txid) {
        System.out.println("接收到checkpoint txid：" + txid);
        this.checkpointTxid = txid;
    }
    public long getCheckpointTxid() {
        return checkpointTxid;
    }
    /**
     * 将checkpoint txid保存到磁盘上去
     */
    public void saveCheckpointTxid() {
        String path = env.getEditlogDirPath() + "/checkpoint-txid.meta";

        RandomAccessFile raf = null;
        FileOutputStream out = null;
        FileChannel channel = null;

        try {
            File file = new File(path);
            if (file.exists()) {
                file.delete();
            }

            ByteBuffer buffer = ByteBuffer.wrap(String.valueOf(checkpointTxid).getBytes());

            raf = new RandomAccessFile(path, "rw");
            out = new FileOutputStream(raf.getFD());
            channel = out.getChannel();

            channel.write(buffer);
            channel.force(false);
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            try {
                if (out != null) {
                    out.close();
                }
                if (raf != null) {
                    raf.close();
                }
                if (channel != null) {
                    channel.close();
                }
            } catch (Exception e2) {
                e2.printStackTrace();
            }
        }
    }
}
